package com.spring.component.rabbitmq.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;
import com.spring.component.redis.RedisUtils;
import com.spring.component.utils.ThreadPoolUtil;

/**
 * 消费者，手动ack
 * 
 * @author liushihua
 * @since 2018年04月25日
 */
@Component
public class Receiver implements ChannelAwareMessageListener {

  private final Logger log = LoggerFactory.getLogger(this.getClass());
  @Autowired
  private ThreadPoolUtil threadPoolUtil;
  @Autowired
  private RedisUtils redisUtils;

  @Override
  public void onMessage(Message message, Channel channel) throws Exception {
    String busMsg = new String(message.getBody());
    log.info("接收消息----{}", busMsg);

    log.info("-----------------------线程数：{}", threadPoolUtil.getExecutor().getPoolSize());
    try {
      threadPoolUtil.acquire();
      threadPoolUtil.getExecutor().execute(() -> {
        try {
          log.info("线程{}工作中...", Thread.currentThread().getName());
          Thread.sleep(300);
          log.info("开门计数----{}", redisUtils.generate("count"));
          log.info("线程{}工作完成!!!", Thread.currentThread().getName());
          // 手动应答ack
          channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
          e.printStackTrace();
        } finally {
          threadPoolUtil.release();
        }
      });
    } catch (Exception e) {
      log.error("接收mq信息出错{}", e);
      channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    }
  }

}
